package com.zcy;

import com.espertech.esper.common.client.EventBean;
import com.espertech.esper.runtime.client.EPEventService;
import com.zcy.event.MyEvent;
import com.zcy.util.EsperBuilder;
import com.zcy.util.TimeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * 14.3.8. Time-Accumulating Window (time_accum or win:time_accum)
 * 类似于FLink当中的SessionWindow
 * <p>
 * 官网描述：
 * 此数据窗口是专用的移动（滑动）时间窗口，与常规时间窗口不同，它会累积事件，
 * 直到在给定时间间隔内没有更多事件到达为止，然后才将累积的事件作为删除流释放。
 * <p>
 * 简单描述就是：流的删除是依据一次间隔没有事件到达，但是事件的触发是实时的
 */
@Slf4j
public class TimeAccumulatingWindow extends TimeUtils {

    private static final String _TimeAcc = "@name('time_acc') select sum(num) from MyEvent#time_accum(5 sec)";

    /**
     * 流的删除是依据一个period没有事件到达，但是流的触发是实时的，这点需要注意
     * sum is : 10
     * sum is : 30
     * sum is : null
     */
    @Test
    public void timeAccumulatingWindowTest() {
        EsperBuilder esperBuilder = new EsperBuilder(_TimeAcc, new Class[]{MyEvent.class});

        esperBuilder.startCompile();

        EPEventService epEventService = esperBuilder.startDeploy((eventBeans, eventBeans1, epStatement, epRuntime) -> {
            for (int i = 0; i < eventBeans.length; i++) {
                EventBean eventBean = eventBeans[i];
                log.info("sum is : {}", eventBean.get("sum(num)"));
            }
        });

        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("001").name("001-name").num(10).build(), MyEvent.class.getSimpleName());
        sleep(TimeUnit.SECONDS, 1L);
        epEventService.sendEventBean(MyEvent.builder().timestamp(System.currentTimeMillis()).amount("002").name("002-name").num(20).build(), MyEvent.class.getSimpleName());

        sleep(TimeUnit.SECONDS, 40L);
    }

}
